package com.alisoft.nano.seda.monitor;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;

import com.alisoft.nano.seda.Event;
import com.alisoft.nano.seda.Stage;
import com.alisoft.nano.seda.notification.StageNotificationListener;

/**
 * Monitor the stage and generate some runtime info and control the stage's performance.
 * 
 * @author zhimin.chenzm, anping.yinap
 *
 * @param <T>
 */
@SuppressWarnings("serial")
@ManagedResource
public class DefaultStageMonitor<T> extends StageMonitor<T> {
	
    private int interval = 10;
	private int expectedProcessingTime = 10;
	private int eventCountToComputeCPUTime = 100;
	
	
	
	private AtomicLong eventCompletedCount = new AtomicLong();
	
	private long lastEventCompletedCount = 0;
	private AtomicLong pendingEventCount = new AtomicLong(); 
	
	private double tps = 0;
	private double maxTPS = 0;
	private int activeThreadsOnMaxTPS = 0;
	
	private AtomicLong totalRunTime = new AtomicLong(0);
	private AtomicLong totalCPUTime = new AtomicLong(0);
	
	private BlockingQueue<Event<T>> queue = new LinkedBlockingQueue<Event<T>>();
	
	private ScheduledExecutorService service;
	
	private Stage<T> stage;
		
	@ManagedAttribute(description="Monitor transaction per second")
	public double getTPS() {
		return tps;
	}
	
	@ManagedAttribute
	public double getForecastProcessingTime() {	
		if (pendingEventCount.get() == 0) {
			return 0;
		}
		if (maxTPS < 1e-9) {
			return Double.MAX_VALUE;
		}
		return getPendingEventCount() / maxTPS * 1.5;
	}

	@ManagedAttribute
	public long getPendingEventCount() {		
		return this.pendingEventCount.get();
	}

	@ManagedAttribute
	public long getEventCompletedCount() {
		return eventCompletedCount.get();
	}
	
	@ManagedAttribute
	public double getCPUTimeRate() {
	    return this.totalCPUTime.doubleValue() / this.totalRunTime.doubleValue();
	}
	
	@ManagedAttribute
	public double getCPUTime() {
	    return this.totalCPUTime.doubleValue() / this.queue.size();
	}
	
	
	public void startProfile() {
		// TODO Auto-generated method stub
		
	}

	public ProfileResult stopProfile() {
		// TODO Auto-generated method stub
		return null;
	}


	/**
	 * In seconds
	 * 
	 * @param interval
	 */
	@ManagedAttribute
	public void setInterval(int interval) {
		this.interval = interval;
	}
	
	@ManagedAttribute
	public int getInterval() {
		return interval;
	}
	
	@ManagedAttribute
	public void setExpectedProcessingTime(int expectedProcessingTime) {
		this.expectedProcessingTime = expectedProcessingTime;
	}

	@ManagedAttribute
	public int getExpectedProcessingTime() {
		return expectedProcessingTime;
	}
	
	@ManagedAttribute
	public double getActiveThreadsOnMaxTPS() {		
		return this.activeThreadsOnMaxTPS;
	}

	@ManagedAttribute
	public double getMaxTPS() {
		return this.maxTPS;
	}
	
	@Override
	protected void onStarted(Stage<T> stage) {	
		this.eventCompletedCount.set(0);
		this.service = Executors.newSingleThreadScheduledExecutor();
		
		this.stage = stage;
		
		this.service.scheduleAtFixedRate(new Runnable() {			
			public void run() {
				long currentCount = eventCompletedCount.get();
				tps = (currentCount - lastEventCompletedCount) / interval;
				lastEventCompletedCount = currentCount;
				
				if (tps > maxTPS) {
					maxTPS = tps;
					// activeThreadsOnMaxTPS will be changed only if current tps is larger than maxTPS
					activeThreadsOnMaxTPS = DefaultStageMonitor.this.stage.getMaximumPoolSize();
				}				
				
				if (getForecastProcessingTime() > getExpectedProcessingTime() ) {
				    DefaultStageMonitor.this.stage.setMaximumPoolSize(Math.max((int) (activeThreadsOnMaxTPS * 1.2), activeThreadsOnMaxTPS + 2));				    
				}
				
			}}, this.interval, this.interval, TimeUnit.SECONDS);
		
		
	}
	
	@Override
	protected void onStopped(Stage<T> stage) {		
		this.service.shutdown();
		this.stage = null;
	}
	
	@Override
	protected void onDone(Event<T> event) {
		this.eventCompletedCount.incrementAndGet();
		this.pendingEventCount.decrementAndGet();
		
		if (this.queue.size() >= eventCountToComputeCPUTime) {
		    Event<T> e = this.queue.poll();
		    this.totalCPUTime.addAndGet(-e.getUserCPUTime());
		    this.totalRunTime.addAndGet(-e.getRunTime());
		} 
		
		this.queue.add(event);
		this.totalRunTime.addAndGet(event.getRunTime());
		this.totalCPUTime.addAndGet(event.getUserCPUTime());
	}

	@Override
	protected void onAccepted(Event<T> event) {
		this.pendingEventCount.incrementAndGet();
	}

}
